package com.wlz.rocketmq.transactionmessage;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author wlz
 * @date 2022-07-02  4:10 下午
 */
public class TransactionListenerImpl implements TransactionListener {

    private AtomicInteger transactionIndex = new AtomicInteger(0);

    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

    //在提交完事务消息后执行。
    //返回COMMIT_MESSAGE状态的消息会立即被消费者消费到。
    //返回ROLLBACK_MESSAGE状态的消息会被丢弃。
    //返回UNKNOWN状态的消息会由Broker过一段时间再来回查事务的状态。
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object o) {
        String tags = msg.getTags();
        //TagA的消息会立即被消费者消费到
        if (StringUtils.contains(tags, "TagA")) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        // TagB的消息会被丢弃
        else if (StringUtils.contains(tags, "TagB")) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
        //其他消息会等待Broker进行事务状态回查。
        else {
            return LocalTransactionState.UNKNOW;
        }
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        String tags = msg.getTags();
        //TagC的消息过一段时间会被消费者消费到
        if (StringUtils.contains(tags, "TagC")) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        //TagD的消息也会在状态回查时被丢弃掉
        else if (StringUtils.contains(tags, "TagD")) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
        //剩下TagE的消息会在多次状态回查后最终丢弃 ( 默认 15次)
        else {
            return LocalTransactionState.UNKNOW;
        }
    }
}
